package com.sg.java.apps;

import com.google.common.collect.Lists;
import com.sg.java.HBaseNamespace;
import com.sg.java.PropertiesUtil;
import com.sg.java.ResourcePath;
import com.sg.java.util.DateUtils;
import com.sg.java.util.HBaseUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Triple;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;

import java.io.IOException;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.text.SimpleDateFormat;
import java.time.*;
import java.time.format.DateTimeFormatter;
import java.time.temporal.ChronoField;
import java.time.temporal.TemporalAccessor;
import java.util.*;

public class GetDataFromHBase {

    private static final Logger log = LoggerFactory.getLogger(GetDataFromHBase.class);

    private static final SimpleDateFormat static_yyyyMMdd_sdf = new SimpleDateFormat("yyyyMMdd");

    private static final String number_length_8_regex = "^\\d{8}$";

    public static final String double_regex = "^(([0])|([1-9]+[0-9]*.{1}[0-9]+)|([0].{1}[1-9]+[0-9]*)|([1-9][0-9]*)|([0][.][0-9]+[1-9]+))$";

    public static void main(String[] args) throws Exception {
        log.info("重庆客户计算程序:从HBase输出数据进行计算输入PostgreSQL");
        if (args == null || args.length == 0) {
            log.error("使用参数:<DATA_DATE>(nonnull eg:20220718) <pageSize>(nullable default 500000 eg:500000)");
            return;
        }
        DateTimeFormatter sdf = DateTimeFormatter.ofPattern("yyyyMMdd");

        String dateStr = args[0];
        if (dateStr.length() != 8) {
            log.error("DATA_DATE长度为8位");
            return;
        }
        String yearMonthPrefix = dateStr.substring(0, 6);
        log.info("yearMonthPrefix:{}", yearMonthPrefix);
        TemporalAccessor startDate = sdf.parse(dateStr);
        int startMonth = startDate.get(ChronoField.DAY_OF_MONTH);
        log.info("startMonth:{}", startMonth);

        List<String> DATA_DATES = new ArrayList<>(31);
        for (int i = startMonth; i <= 31; i++) {
            String day = String.valueOf(i);
            if (day.length() == 1) {
                day = "0" + day;
            }
            DATA_DATES.add(yearMonthPrefix + day);
        }
//        final String[] DATA_DATES = args[0].split(",");
        StringJoiner dd_sj = new StringJoiner("\t");
        for (String data_date : DATA_DATES) {
            dd_sj.add(data_date);
        }
        log.info("同步指定日期数据：{}", dd_sj);

//        final String DATA_DATE_errorMsg = "DATA_DATE为8位数字，年4位月2位日2位，yyyyMMdd";

        Class.forName("org.postgresql.Driver");

        final Properties pgProp = PropertiesUtil.createPropertiesFromResource(ResourcePath.dws_business_jdbc_properties);
        final String url = pgProp.getProperty("url");
        final String user = pgProp.getProperty("user");
        final String password = pgProp.getProperty("password");
        log.info("读取业务中台pg库连接配置：{}", pgProp);
        final int limit = args.length == 2 ? Integer.parseInt(args[1]) : 500000;
        log.info("设置每轮查询数据条数：{}", limit);
        final long global_s = System.currentTimeMillis();
        log.info("开始时间戳：{}", global_s);
        final Connection hBaseConn = HBaseUtils.getHBaseConn(PropertiesUtil.createPropertiesFromResource(ResourcePath.hbase_properties));
        log.info("创建全局唯一hbase连接");
        DATA_DATES.forEach(DATA_DATE -> {
//            if (DATA_DATE.length() != 8 || !DATA_DATE.matches(number_length_8_regex)) {
//                log.error(DATA_DATE_errorMsg);
//                throw new RuntimeException(DATA_DATE_errorMsg);
//            }
            log.info("查询数据日期：{}", DATA_DATE);
            final String yyyy = DATA_DATE.substring(0, 4);
            final String MM = DATA_DATE.substring(4, 6);
            final String dd = DATA_DATE.substring(6, 8);
            String DATA_DATE_yyyyMMdd = yyyy + "-" + MM + "-" + dd;

//        final String csvPathPrefix = "/qkjdy/" + DATA_DATE + "_";
            final List<Triple<String, String, String>> triples = Lists.newArrayList(
                    Triple.of(HBaseNamespace.CQ_HBASE_NAMESPACE + ":cms_volt_curve_" + DATA_DATE, "pms3_qkjdy.gddy_amr_volt_curve", "U"),
                    Triple.of(HBaseNamespace.CQ_HBASE_NAMESPACE + ":cms_cur_curve_" + DATA_DATE, "pms3_qkjdy.gddy_amr_cur_curve", "I"),
                    Triple.of(HBaseNamespace.CQ_HBASE_NAMESPACE + ":cms_power_curve_" + DATA_DATE, "pms3_qkjdy.gddy_amr_power_curve", "P")
            );
            triples.forEach(triple -> {
                final TableName tableName = TableName.valueOf(triple.getLeft());
                try (Admin admin = hBaseConn.getAdmin()) {
                    final boolean tableAvailable = admin.isTableAvailable(tableName);
                    if (!tableAvailable) {
                        log.info(tableName + "不存在，数据日期：{}", DATA_DATE);
                        return;
                    }
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
                log.info(triple.getLeft() + "\t获取hbase表：{}", tableName);
                String finalSql = null;
                try (
//                        final CSVPrinter csvPrinter = CsvUtils.newDefaultCSVPrinter(csvPathPrefix + pair.getLeft() + ".csv");
                        final HTable table = (HTable) hBaseConn.getTable(tableName);
                        final java.sql.Connection pgConn = DriverManager.getConnection(url, user, password)
                ) {
                    int round = 0;
                    int realTimeSize;
                    int totalSize = 0;
                    long totalAffectRows = 0;
                    Result next = null;
                    StringJoiner sj_96 = new StringJoiner(",");
                    //P->p I->i U->u
                    String lc = triple.getRight().toLowerCase();
                    for (int i = 1; i <= 96; i++) {
                        sj_96.add(lc + i);
                    }
                    //todo
//                    pgConn.prepareStatement("create table if not exists "+triple.getMiddle()+" ");
                    String unique;
                    if (triple.getRight().equals("U") || triple.getRight().equals("I")) {
                        unique = "phase_flag";
                    } else {
                        unique = "data_type";//P
                    }
                    String preSql = "insert into " + triple.getMiddle() + " (id,data_date,org_no," + unique + "," + sj_96 + ") values ";
                    log.info(triple.getLeft() + "\tpreSql：{}", preSql);
                    do {
                        long local_s = System.currentTimeMillis();
                        round++;
                        realTimeSize = 0;
                        Scan scan = new Scan();
                        scan.addFamily(Bytes.toBytes("info"));
                        scan.addFamily(Bytes.toBytes(triple.getRight()));
//                        FilterList filterList = new FilterList();
                        scan.setLimit(limit);
//                        filterList.addFilter(new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("DATA_DATE"), CompareOperator.EQUAL, Bytes.toBytes(DATA_DATE)));
//                        scan.setFilter(filterList);
                        if (next != null) {
                            log.info(triple.getLeft() + "\t下次分页数据起始行：{}", Bytes.toString(next.getRow()));
                            scan.withStartRow(next.getRow());
                        }
                        ResultScanner scanner = table.getScanner(scan);
                        Map<Integer, String> map = new HashMap<>(96);
                        StringJoiner valuesSj = new StringJoiner(",");
                        final Jedis jedis = new Jedis("25.64.41.53", 6379, 100000);
                        jedis.auth("Huawei@169");
                        jedis.select(0);
                        for (Result result : scanner) {
                            realTimeSize++;
                            totalSize++;
                            next = result;
                            List<Cell> cells = next.listCells();
                            String METER_ID = Bytes.toString(next.getValue(Bytes.toBytes("info"), Bytes.toBytes("METER_ID")));
                            String key = "PBGX@" + METER_ID;
                            String relation = jedis.get(key);
                            if (relation != null && relation.endsWith("02")) {
                                StringJoiner valueSj = new StringJoiner(",", "(", ")");
//                            String DATA_DATE = Bytes.toString(next.getValue(Bytes.toBytes("info"), Bytes.toBytes("DATA_DATE")));
                                String uniqueValue = Bytes.toString(next.getValue(Bytes.toBytes("info"), Bytes.toBytes(unique.toUpperCase())));
                                String ORG_NO = Bytes.toString(next.getValue(Bytes.toBytes("info"), Bytes.toBytes("ORG_NO")));
                                valueSj.add("'" + METER_ID + "'");
                                valueSj.add("'" + DATA_DATE_yyyyMMdd + "'");
                                valueSj.add("'" + ORG_NO + "'");
                                valueSj.add("'" + uniqueValue + "'");
                                cells.stream()
                                        .filter(cell -> Bytes.toString(CellUtil.cloneFamily(cell)).equals(triple.getRight()))
                                        .forEach(cell -> map.put(DateUtils.getPoint(Bytes.toString(CellUtil.cloneQualifier(cell))), Bytes.toString(CellUtil.cloneValue(cell))));
                                for (int i = 1; i <= 96; i++) {
                                    String value = map.getOrDefault(i, "null");
                                    //value 可能为空字符串
                                    if (StringUtils.isBlank(value) || !value.matches(double_regex)) {
                                        value = "null";
                                    }
                                    valueSj.add(value);
                                }
                                valuesSj.add(valueSj.toString());
                                map.clear();
                            }
//                            csvPrinter.printRecord(sj.toString());
                        }
                        jedis.close();
                        if (valuesSj.length() == 0) {
                            log.info(triple.getLeft() + "\t此轮无可插入pg库数据，跳过");
                            continue;
                        }
                        String postSql = valuesSj.toString();
                        finalSql = preSql + postSql;
                        PreparedStatement ps = pgConn.prepareStatement(finalSql);
                        int affectRows = ps.executeUpdate();
                        totalAffectRows += affectRows;
                        log.info(triple.getLeft() + "\t此轮轮数：{}\t此轮hbase读取数据条数：{}\t此轮pg库写入配变数据条数：{}\t此轮耗时：{}s", round, realTimeSize, affectRows, (System.currentTimeMillis() - local_s) / 1000);
                    } while (realTimeSize >= limit);
                    log.info(triple.getLeft() + "\t总轮数：{}\t总hbase读取数据条数：{}\t总pg库写入配变数据条数：{}\t总耗时：{}s", round, totalSize, totalAffectRows, (System.currentTimeMillis() - global_s) / 1000);
                } catch (Exception e) {
                    e.printStackTrace();
                    log.error("finalSql：{}", finalSql);
                    throw new RuntimeException(e);
                }
            });
        });

    }

    public static int getPoint(Date time) {
        Instant instant = time.toInstant();
        ZoneId zone = ZoneId.systemDefault();
        //Date转换为LocalDateTime
        LocalDateTime localDateTime = LocalDateTime.ofInstant(instant, zone);
        //获取当前时间毫秒值
        long beginTime = localDateTime.toInstant(ZoneOffset.of("+8")).toEpochMilli();
        //获取当天初始时间毫秒值
        long endTime = LocalDateTime.of(localDateTime.toLocalDate(), LocalTime.MIN).toInstant(ZoneOffset.of("+8")).toEpochMilli();
        //计算对应点数
        long l = (beginTime - endTime) / (60000 * 15);
//        System.out.println(l);
        if (l == 0) l = 96;
        return (int) l;
    }

}
